import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.ApplicationProtocolNames;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.File;
import java.nio.charset.Charset;
import java.util.List;

public class ETCD {
    private static final String endpoints = "https://192.168.1.152:2379,https://192.168.1.153:2379,https://192.168.1.155:2379";
    //etcl客户端链接
    private static Client client = null;

    public static synchronized Client getEtclClient() throws Exception {
        File CA = new File("C:\\JAVAProject\\SSLTest\\src\\main\\resources\\ca-2.crt");

        if (null == client) {

            ApplicationProtocolConfig apn = new ApplicationProtocolConfig(
                    ApplicationProtocolConfig.Protocol.ALPN,
                    ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
                    ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
                    ApplicationProtocolNames.HTTP_2,
                    ApplicationProtocolNames.HTTP_1_1);

            SslContext sslContext = SslContextBuilder.forClient().applicationProtocolConfig(apn).trustManager(CA).build();
            client = Client.builder().sslContext(sslContext).endpoints(endpoints.split(",")).build();

        }
        return client;
    }

    /**
     * etcd的监听，监听指定的key，当key 发生变化后，监听自动感知到变化。
     *
     * @param key 指定需要监听的key
     */
    public static void initListen(Client etcdClient, String key, SourceFunction.SourceContext<String> ctx) {
        try {
            //加载初始值
            getInitValue(etcdClient.getKVClient().get(ByteSequence.from(key, Charset.defaultCharset())).get().getKvs());

            Watch.Watcher watcher = null;
            try {
                MyConsumer consumer = new MyConsumer(ctx);
                etcdClient.getWatchClient().watch(ByteSequence.from(key, Charset.defaultCharset()), Watch.listener(consumer));
            } catch (Exception e) {
                e.printStackTrace();
            }

        } catch (Exception e) {
            System.out.println("etcd listen start cause Exception:{" + e + "}");
        }

    }

    /**
     * etcd的监听，监听指定的key，当key 发生变化后，监听自动感知到变化。
     *
     * @param key 指定需要监听的key
     */
    public static void initListen(Client etcdClient, String key) {
        try {
            //加载初始值
            getInitValue(etcdClient.getKVClient().get(ByteSequence.from(key, Charset.defaultCharset())).get().getKvs());

            Watch.Watcher watcher = null;
            try {
                MyConsumer consumer = new MyConsumer();
                etcdClient.getWatchClient().watch(ByteSequence.from(key, Charset.defaultCharset()), Watch.listener(consumer));
            } catch (Exception e) {
                e.printStackTrace();
            }

        } catch (Exception e) {
            System.out.println("etcd listen start cause Exception:{" + e + "}");
        }

    }

    private static String getInitValue(List<KeyValue> kvs) {
        if (kvs.size() > 0) {
            String config = kvs.get(0).getKey().toString(Charset.defaultCharset());
            String value = kvs.get(0).getValue().toString(Charset.defaultCharset());
            System.out.println("etcd 's config 's config key is :{" + config + "},value is:{" + value + "}");
            return value;
        } else {
            return null;
        }
    }
}
